import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/11/21 13:20
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object modify_yushou {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    val brand_data = sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/shuang11/join_data_yushou_v2_modify",
      "s3a://o2o-dataproces-group/panzonghao/shuang11/nojoin_data_yushou_modify").drop("is_showLive","fans")
      .where("cast(sellCount as bigint)>0 and cast(salesAmount as double)>0 and cast(priceText as double)>0")


    brand_data.createOrReplaceTempView("brand_data")
    //添加直播
    val data1 = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/tmall_live_detail_path")
    val data2 = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/shuang11/normalSales/tmall/tmall_live_id_path")
    data1.createOrReplaceTempView("data1")
    data2.createOrReplaceTempView("data2")

    val data3 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.good_id as good_ids
         |from data2 t1
         |inner join
         |data1 t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    data3.createOrReplaceTempView("data3")
    val data4 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |cast(t1.sellCount as bigint) as sellCounts,
         |t2.viewerCount,
         |t2.roomType,
         |t2.liveId,
         |t2.anchorId,
         |t2.liveTime,
         |t2.roomTypeName,
         |t2.nick,
         |t2.liveTitle,
         |t2.commodityCount,
         |case
         |when t2.good_id is null then 'false'
         |else 'true'
         |end as is_showLive
         |from brand_data t1
         |left join
         |data3 t2
         |on t1.good_id=t2.good_id
       """.stripMargin).drop("sellCount").withColumnRenamed("sellCounts","sellCount")

    data4.dropDuplicates("good_id").repartition(1).write.orc("s3a://o2o-dataproces-group/panzonghao/modify_shuang11")

  }
}
